package com.oracle.calllog.analyze.runner;

import com.oracle.calllog.analyze.model.ContactDimBean;
import com.oracle.calllog.analyze.outputformat.MyDBOutputFormat;
import com.oracle.calllog.analyze.reducer.CallLogAnalyzeReducer;
import com.oracle.calllog.analyze.model.CountDuration;
import com.oracle.calllog.analyze.util.Constants;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;

import com.oracle.calllog.analyze.mapper.CallLogAnalyzeMapper;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class CallLogAnalyzeRunner implements Tool {
	private Configuration conf;

	@Override
	public void setConf(Configuration conf) {
		this.conf = conf;
	}

	@Override
	public Configuration getConf() {
		return this.conf;
	}

	@Override
	public int run(String[] strings) throws Exception {
		// 获取流水线作业对象
		Job job = Job.getInstance(this.conf, "Calllog_Analyze");

		job.setJarByClass(CallLogAnalyzeRunner.class);

		// 创建一个用于对HBase进行扫描的描述对象
		Scan scan = new Scan();
		// 此需求只需扫描联系人的姓名那一列即可，其余信息在RowKey中可以找到
		scan.addColumn(Bytes.toBytes(Constants.FAMILY_CALL),Bytes.toBytes(Constants.QUALIFIER_CONTACT1_NAME));
		// 设置Mapper阶段，会自动初始化InputForamt
		TableMapReduceUtil.initTableMapperJob(Constants.TABLE_NAME, scan, CallLogAnalyzeMapper.class, Text.class, Text.class, job);

		// 设置Reducer阶段，并设置自定义Outputformat
		job.setReducerClass(CallLogAnalyzeReducer.class);
		job.setOutputKeyClass(ContactDimBean.class);
		job.setOutputValueClass(CountDuration.class);
		job.setOutputFormatClass(MyDBOutputFormat.class);

//		//测试一下文本格式的输出是否有问题：如果输出路径已存在，则先删除此路径
//		Path outPath = new Path(strings[0]);
//		FileSystem fs = FileSystem.get(conf);
//		if(fs.exists(outPath)) {
//			fs.delete(outPath, true);
//		}
//		FileOutputFormat.setOutputPath(job, outPath);

		//提交
		boolean result = job.waitForCompletion(true);
		return result?0:1;
	}


	public static void main(String[] args)  {
		//不需要从外界传入路径参数了，程序里已经处理完了
		try {
			ToolRunner.run(new CallLogAnalyzeRunner(), args);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}
